package com.practice.gmall.realtime.app.dwd;

import com.practice.gmall.realtime.app.BaseSQLApp;
import com.practice.gmall.realtime.util.MyKafkaUtil;
import com.practice.gmall.realtime.util.MySQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: DWD_07_TradeOrderRefound
 * Package: com.practice.gmall.realtime.app.dwd
 * Description:
 * <p>
 * 交易域退款事实表
 *      粒度：商品退款粒度
 *      维度：用户、商品、省份、退款时间
 *      度量：退款金额
 * <p>
 * 使用到的表
 *      refund_payment base_dic order_detail order_activity_detail order_coupon_detail
 * <p>
 * 使用到的API
 *      cdc   Flink SQL lookup    Flink SQL join
 * @Author lzy.ethan@qq.com
 * @Create 2023-02-15 8:57
 */
public class DWD_07_TradeOrderRefundSuccess extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_07_TradeOrderRefundSuccess().init(3338,"DWD_07_TradeOrderRefund",2);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // read ods_db
        MyKafkaUtil.readTopicOdsDb(tEnv,"DWD_07_TradeOrderRefund");

        // filter refund line from order_refund_info
        Table refundPayment = tEnv.sqlQuery("" +
                "SELECT" +
                "    `data`['order_id'] order_id," +
                "    `data`['sku_id'] sku_id," +
                "    `data`['payment_type'] payment_type," +
                "    `data`['total_amount'] total_amount," +
                "    `data`['create_time'] create_time," +
                "    `data`['callback_time'] date_id," +
                "    pt ," +
                "    ts " +
                "FROM ods_db " +
                "WHERE" +
                "    `database` = 'gmall2022'" +
                "    AND `table` = 'refund_payment'" +
                "    AND  `type` = 'insert'");
        tEnv.createTemporaryView("refund_payment",refundPayment);

        // read base_dic
        tEnv.executeSql(MySQLUtil.getBaseDicLookUp());


        // create table order_detail
        tEnv.executeSql("" +
                "CREATE TABLE dwd_order_detail(  " +
                "    id  string,     " +
                "    order_id  string,     " +
                "    user_id  string,    " +
                "    sku_id  string,     " +
                "    sku_name  string,     " +
                "    order_price  string,     " +
                "    sku_num  string,     " +
                "    date_id  string,     " +
                "    province_id  string,  " +
                "    source_type_code  string,  " +
                "    source_type_name  string,  " +
                "    source_id  string,  " +
                "    activity_id  string,  " +
                "    activity_rule_id   string,  " +
                "    coupon_id  string,  " +
                "    split_original_amount  string,     " +
                "    split_total_amount   string,  " +
                "    split_activity_amount   string,  " +
                "    split_coupon_amount   string,  " +
                "    `ts` BIGINT , " +
                "     et as to_timestamp_ltz(ts,0),   " +
                "     watermark for et as et - interval '3' second  " +
                ") " +
                MyKafkaUtil.getKafkaSourceDDL("dwd_order_detail", "DWD_07_TradeOrderRefund"));


        // join query
        Table refundSuccess = tEnv.sqlQuery("" +
                "SELECT " +
                "    dod.id , " +
                "    dod.order_id , " +
                "    dod.user_id , " +
                "    dod.sku_id , " +
                "    dod.sku_name , " +
                "    dod.order_price , " +
                "    dod.sku_num , " +
                "    date_format(rp.date_id ,'yyyy-MM-dd') date_id , " +
                "    dod.province_id , " +
                "    dod.source_type_code , " +
                "    dod.source_type_name , " +
                "    dod.source_id , " +
                "    rp.payment_type , " +
                "    dic.dic_name payment_type_name, " +
                "    dod.activity_id , " +
                "    dod.activity_rule_id  , " +
                "    dod.coupon_id , " +
                "    dod.split_original_amount , " +
                "    dod.split_total_amount  , " +
                "    dod.split_activity_amount  , " +
                "    dod.split_coupon_amount  , " +
                "    rp.total_amount refund_total_amount, " +
                "    rp.ts " +
                "FROM dwd_order_detail dod JOIN refund_payment rp on dod.order_id = rp.order_id  " +
                "        AND dod.sku_id = rp.sku_id " +
                "    JOIN base_dic FOR SYSTEM_TIME AS OF rp.pt AS dic on dic.dic_code = rp.payment_type");


        // create a table sink table
        tEnv.executeSql("" +
                "CREATE TABLE dwd_refund_success( " +
                "    id STRING, " +
                "    order_id STRING, " +
                "    user_id STRING, " +
                "    sku_id STRING, " +
                "    sku_name STRING, " +
                "    order_price STRING, " +
                "    sku_num STRING, " +
                "    date_id STRING, " +
                "    province_id STRING, " +
                "    source_type_code STRING, " +
                "    source_type_name STRING, " +
                "    source_id STRING, " +
                "    payment_type STRING, " +
                "    payment_type_name STRING, " +
                "    activity_id STRING, " +
                "    activity_rule_id  STRING, " +
                "    coupon_id STRING, " +
                "    split_original_amount STRING, " +
                "    split_total_amount  STRING, " +
                "    split_activity_amount  STRING, " +
                "    split_coupon_amount  STRING, " +
                "    refund_total_amount STRING, " +
                "    ts BIGINT)" +
                MyKafkaUtil.getKafkaSinkDDL("dwd_refund_success"));

        // write to sink table
        tEnv.executeSql("insert into dwd_refund_success select * from " + refundSuccess);

    }
}
